// Copyright (c) 2021 Presto Labs Pte. Ltd.
// Author: gekim

#pragma once

#include <memory>
#include <string>
#include <unordered_map>
#include <utility>

#include <boost/date_time/gregorian/gregorian.hpp>
#include <fmt/format.h>

// internal
#include "coin/feed/fastfeed/feed.pb.h"
#include "coin2/base/time.h"
#include "coin2/exchange/base/market/market_exchange_api.h"
#include "coin2/feed/fastfeed/writer.h"
#include "orio/io/base.h"
#include "orio/io/concat.h"
#include "orio/io/file.h"
#include "orio/record/simple.h"

namespace coin2::feed::fastfeed {
namespace impl {

using ::coin2::exchange::base::market::MarketExchangeApi;
using ::fastfeed::proto::Feed;

class FastFeedUniverseWriter : public FastFeedWriter {
 public:
  FastFeedUniverseWriter(
      const std::string& root_dir,  // may or may not end with '/'
      const std::string& machine,
      const MarketExchangeApi& mea,
      const std::string& recipe,
      int worker,
      const std::string& extension,
      const std::string& start,
      const std::string& dest_proto_tag)
      : FastFeedWriter(root_dir, machine, mea, recipe, worker, extension, start),
        dest_proto_tag_(dest_proto_tag) {
  }

  ~FastFeedUniverseWriter() {
    if (writer_) {
      writer_->Flush();
    }
  }

  void Write(Feed* in) override {
    // auto& symbol = in->symbol();
    int64_t date = in->timestamp() / (24LL * 60 * 60 * 1'000'000'000);
    if (!writer_ || last_dates_.at(dest_proto_tag_) != date) {
      // {config.fastfeed_root_dir}/{machine}/date/{mea}/
      // '%s.W%s.%s' % (str(product), worker extension)

      last_dates_[dest_proto_tag_] = date;
      auto ptime = PtimeFromTimestamp(in->timestamp());
      auto YYYYMMDD = boost::gregorian::to_iso_string(ptime.date());
      std::experimental::filesystem::path path = fmt::format(
          "{root_dir}/{machine}/{YYYYMMDD}/{mea}.{recipe}/"
          "{dest_proto_tag}{start_time_tag}.W{worker}{dot_extension}",
          fmt::arg("root_dir", root_dir_),
          fmt::arg("machine", machine_),
          fmt::arg("YYYYMMDD", YYYYMMDD),
          fmt::arg("start_time", start_time_),
          fmt::arg("mea", mea_.String()),
          fmt::arg("recipe", recipe_),
          fmt::arg("dest_proto_tag", dest_proto_tag_),
          fmt::arg("start_time_tag", start_time_),
          fmt::arg("worker", worker_),
          fmt::arg("dot_extension", extension_));

      std::string tmp_name = fmt::format(
          "coin2_fastfeed/{machine}/{YYYYMMDD}/{mea}.{recipe}/"
          "{dest_proto_tag}{start_time_tag}.W{worker}{dot_extension}",
          fmt::arg("machine", machine_),
          fmt::arg("YYYYMMDD", YYYYMMDD),
          fmt::arg("start_time", start_time_),
          fmt::arg("mea", mea_.String()),
          fmt::arg("recipe", recipe_),
          fmt::arg("dest_proto_tag", dest_proto_tag_),
          fmt::arg("start_time_tag", start_time_),
          fmt::arg("worker", worker_),
          fmt::arg("dot_extension", extension_));
      auto tmp_path = orio::io::GetTemporaryFile(tmp_name);

      if (!writer_) {
        writer_ = std::unique_ptr<orio::io::Writer<orio::record::SimpleRecord>>(
            orio::record::SimpleWriter::ToFile(tmp_path, path));
      } else {
        writer_.reset(orio::record::SimpleWriter::ToFile(tmp_path, path));
      }

      LOG(INFO) << "[FastFeedUniverseWriter] creating a fastfeed file\n"
                << "dst file: " << path.c_str() << "\ntmp file: " << tmp_path;
    }

    in->SerializeToString(&buf1_);
    buf2_.resize(buf1_.size() + sizeof(uint32_t));
    auto* record = reinterpret_cast<orio::record::SimpleRecord*>(buf2_.data());
    record->len = buf1_.size();
    std::memcpy(&record->data[0], buf1_.data(), record->len); 
    writer_->Write(*record);
  }

 private:
  std::unique_ptr<orio::io::Writer<orio::record::SimpleRecord>> writer_;
  std::string dest_proto_tag_;
};

}  // namespace impl

using impl::FastFeedUniverseWriter;

}  // namespace coin2::feed::fastfeed
